#include "BDZmqReceiver.h"

#include <iostream>

using namespace std;

BDZmqReceiver::BDZmqReceiver(std::string address)
	: m_sAddress(address)
	, m_pSocket(NULL)
{
	
}


BDZmqReceiver::~BDZmqReceiver()
{

}

int BDZmqReceiver::Init()
{
	m_pSocket = zmq_socket(m_pContext, ZMQ_PULL);
	if (m_pSocket == NULL)
	{
		return -1;
	}

	int rc = zmq_connect(m_pSocket, m_sAddress.c_str());
	if (rc != 0)
	{
		return -1;
	}
	m_PollItem.socket = m_pSocket;
	m_PollItem.events = ZMQ_POLLIN;
	return 0;
}

int BDZmqReceiver::DeInit()
{
	int rc = zmq_close(m_pSocket);
	if (rc != 0)
	{
		return -1;
	}
	return 0;
}

int BDZmqReceiver::RecvMsg(void* buffer, unsigned int length)
{
	size_t src_len = 0;
	int rc = zmq_poll(&m_PollItem, 1, -1);
	if (rc < 0)
	{
		return -1;
	}
	else if (rc == 0)
	{
		return 0;
	}
	else
	{
		zmq_msg_t msg;
		rc = zmq_msg_init(&msg);
		if (rc != 0)
		{
			printf("%s\n", zmq_strerror(errno));
			return -1;
		}
		
		rc = zmq_msg_recv(&msg, m_pSocket, 0);

		if (rc == 0)
		{
			printf("%s\n", zmq_strerror(errno));
			return -1;
		}

		void* src = zmq_msg_data(&msg);
		src_len = zmq_msg_size(&msg);

		if (src_len > length) {
			return -1;
		}
		memcpy(buffer, src, src_len);

		zmq_msg_close(&msg);
	}
	return src_len;
}